package com.zhaosc.spark.sql.udf

import org.apache.spark.sql.expressions.UserDefinedAggregateFunction
import org.apache.spark.sql.expressions.MutableAggregationBuffer
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.LongType
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.DataType
import org.apache.spark.sql.types.DataTypes
import org.apache.spark.sql.types.StructField
import org.apache.commons.lang3.StringUtils

object NameCountAggregateFunction extends UserDefinedAggregateFunction {

  /**
   * 入参类型
   */
  override def inputSchema: StructType = new StructType().add("input", DataTypes.StringType)

  /**
   *
   * 缓存类型
   */
  def bufferSchema: StructType = new StructType().add(StructField("name", DataTypes.StringType));

  /**
   * 返回类型
   */
  def dataType: DataType = DataTypes.StringType;

  /**
   * / 聚合函数是否是幂等的，即相同输入是否总是能得到相同输出
   */
  def deterministic: Boolean = true

  /**
   * 初始化
   */
  def initialize(buffer: MutableAggregationBuffer) = {
    buffer(0) = "";
  }

  /**
   */
  def update(buffer: MutableAggregationBuffer, input: Row) = {
    if (StringUtils.isNotBlank(buffer.getString(0))) {
      buffer.update(0, buffer.getString(0).concat("-").concat(input.getString(0)));
    } else if (StringUtils.isNotBlank(input.getString(0))) {
      buffer.update(0, input.getString(0));
    }
  }

  def merge(buffer: MutableAggregationBuffer, input: Row) = {
    if (StringUtils.isNotBlank(buffer.getString(0))) {
      buffer.update(0, buffer.getString(0).concat("-").concat(input.getString(0)));
    } else {
      buffer.update(0, input.getString(0));
    }
  }

  def evaluate(buffer: Row) = {
    println(buffer.getString(0))
    buffer.getString(0);

  }
}